/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.mycompany.mqdemo3;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *发送消息，接收消息后，更改字段内容转发给新的消息队列
 * @author dell
 */
public class Receive {
    public static final String NAME_SERVER = "192.168.203.128:9879";
    public static final String TOPIC = "spftest11";
    public static final String GROUP_NAME = "group_test_1";
    public static final String TOPIC1 = "spftest22";
    public static String content_string;
    public static int content_int;
    public static boolean content_boolean;
    public static Map[] content_structs=new Map[2];
    static{
      Properties p = new Properties();
        try {
            //类加载时读取配置文件(字段的内容),并转化成相应的类型
           p.load(Receive.class.getClassLoader().getResourceAsStream("config.properties"));
            content_string=p.getProperty("content_string");
            content_int=Integer.parseInt(p.getProperty("content_int"));
            content_boolean=Boolean.parseBoolean(p.getProperty("content_boolean"));
            JSONObject json=JSONObject.parseObject(p.getProperty("content_struct1"));
            Map content_struct1=(Map)json;     
            for(Object str:content_struct1.keySet()){
                String str1=(String)str;
                if(content_struct1.get(str1) instanceof String){
                   content_struct1.put(str1, (String)content_struct1.get(str)); 
                }else{
                   content_struct1.put(str1,new Long((int)content_struct1.get(str))); 
                }       
            }
            content_structs[0]=content_struct1;
            JSONObject json1=JSONObject.parseObject(p.getProperty("content_struct2"));
            Map content_struct2=(Map)json1;
            for(Object str:content_struct2.keySet()){
                String str2=(String)str;
                if(content_struct2.get(str2) instanceof String){
                   content_struct2.put(str2, content_struct2.get(str).toString()); 
                }else{
                   content_struct2.put(str2,new Long((int)content_struct2.get(str))); 
                }       
            }  content_structs[1]=content_struct2;
        } catch (IOException ex) {
            ex.printStackTrace();
        }       
    }
    public static void main(String[] args) throws REConnectionException, InterruptedException {
       //创建连接
       REConnection conn=new REConnection(NAME_SERVER);
       final LinkedBlockingQueue<REMessage> lbq=new LinkedBlockingQueue<>();
       //创建线程r1接收消息，
       Runnable r1=new Runnable(){
           @Override
           public void run() {
               try {
                   REReceiveSessionBuilder builder=(REReceiveSessionBuilder)conn.getReceiveSessionBuilder(TOPIC);
                   builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
                       @Override
                       public boolean handle(REMessage message) {
                           if(message==null){
                               return false;
                           }
                           try {
                           
                               lbq.put(message);
                           } catch (Exception ex) {
                               ex.printStackTrace();
                           }
                           return true;
                       }
                   });
                   builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
                       @Override
                       public boolean handle(byte[] message) {
                           System.out.println(message.toString()+"||"+message.length);
                           return true;
                       }
                   });
                   builder.setGroupName(GROUP_NAME);
                   builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
                   builder.setConsumeThreadNum(10);
                       REReceiveSession session=(REReceiveSession)builder.build();
                       session.start();
               } catch (Exception ex) {
                   ex.printStackTrace();
               }
           }   
       };
       //创建线程r2转发消息
       Runnable r2=new Runnable(){
           @Override
           public void run() {     
               try {     
                   RESendSessionBuilder sendBuilder=(RESendSessionBuilder)conn.getSendSessionBuilder(TOPIC1);
                   //设置字段
                   sendBuilder.addColumn("booleanTest", REFieldType.Boolean, true);
                   sendBuilder.addColumn("intTest", REFieldType.Int, true);
                   sendBuilder.addColumn("stringTest", REFieldType.String, true);
                   sendBuilder.addColumn("structsTest", REFieldType.Structs, true);
                   HashMap<String,REFieldType>map=new HashMap();
                   map.put("str1", REFieldType.String);
                   map.put("long1", REFieldType.Long);
                   REFieldType.Structs.setDetail("structsTest", map);
                   RESendSession session=null;
                   try {
                       session=(RESendSession)sendBuilder.build();
                       session.start();
                   } catch (RESessionException ex) {
                       ex.printStackTrace();
                   } catch (REConnectionException ex) {
                       ex.printStackTrace();
                   }
                   int count=0;
                   //从队列中拿去消息
                   for(REMessage message:lbq){
                       for(Object[]obj:message.data){
                           try {
                               for(int i=0;i<message.colNames.length;i++){
                                   if(obj[i] instanceof String){
                                       session.setString(message.colNames[i],content_string);
                                       System.out.println(message.colNames[i]);
                                   }else if(obj[i] instanceof Boolean){
                                       session.setBoolean(message.colNames[i],content_boolean);
                                       System.out.println(message.colNames[i]);
                                   }else if(obj[i] instanceof Integer){
                                       session.setInt(message.colNames[i], content_int);
                                       System.out.println(message.colNames[i]);
                                   }
                                   else{                           
                                       session.setStructs(message.colNames[i],content_structs);
                                       System.out.println(message.colNames[i]);
                                   }
                                   
                               }
                               session.add();
                           } catch (RESessionException ex) {
                               ex.printStackTrace();
                           } 
                           session.flush();
                               count++;
                               System.out.println("message"+count);
                               Thread.sleep(2000);
                           
                       }
                       
                   }
               } catch (REConnectionException ex) {
                   Logger.getLogger(Receive.class.getName()).log(Level.SEVERE, null, ex);
               } catch (RESessionException ex) {
                   Logger.getLogger(Receive.class.getName()).log(Level.SEVERE, null, ex);
               } catch (InterruptedException ex) {
                   Logger.getLogger(Receive.class.getName()).log(Level.SEVERE, null, ex);
               }
       };  
     };
       //保 证取消息和转发消息的先后顺序
       Thread t1=new Thread(r1);
       Thread t2=new Thread(r2);
       t1.start();
       t1.join();
       t2.start();
   }
}
